package com.my.springboot.tuling.concurrent.demo;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.annotation.JsonAlias;
import org.apache.commons.lang.StringUtils;

import java.io.*;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Random;

/**
 * java多线程读取同一个文件，并多线程插入数据库
 * 当需要读取的文件较大时，使用单线程读取可能需要花费较长时间，而使用多线程可以将文件分成多个块，每个线程负责读取其中的一部分，从而并行读取整个文件，减少了读取时间。
 */
public class MultiThreadReadOneFile {
    private static Connection conn = null;
    private static PreparedStatement ps = null;
    // 文件路径
    private static String filePath = "d:" + File.separator + "gc3" + File.separator + "update.txt";

    private int wirteRows = 10000000;//1000万
    public static void main(String[] args) throws IOException {
        long start = System.currentTimeMillis();
        MultiThreadReadOneFile readFile = new MultiThreadReadOneFile();
        // 写文件数据
        //readFile.writeFile();
        // 读文件数据入库
        readFile.save(start);
    }

    private void save(long start) {
        int threadNum = 5; // 线程数

        try {
            // 加载驱动
            Class.forName("com.mysql.jdbc.Driver");
            // 获取连接
            conn = DriverManager.getConnection("jdbc:mysql://localhost/case_test?serverTimezone=UTC", "root", "root");
            // 设置自动提交为false
            conn.setAutoCommit(false);
            // 创建预编译语句
            ps = conn.prepareStatement("INSERT INTO sys_user (id, name,age) VALUES (?, ?, ?)");

            // 创建线程数组
            Thread[] threads = new Thread[threadNum];
            // 计算每个线程要读取的行数
            BufferedReader br = new BufferedReader(new FileReader(filePath));
            int totalLines = 0;
            while (br.readLine() != null) {
                totalLines++;
            }
            br.close();
            int linesPerThread = totalLines / threadNum;
            System.out.println("行数：" + totalLines + ",每个线程执行行数：" + linesPerThread);
            // 启动线程
            for (int i = 0; i < threadNum; i++) {
                int startLine = i * linesPerThread + 1;
                int endLine = (i == threadNum - 1) ? totalLines : (i + 1) * linesPerThread;
                threads[i] = new Thread(new FileReaderThread(filePath, startLine, endLine, ps,conn, start));
                threads[i].start();
            }

            // 等待所有线程执行完毕
            for (int i = 0; i < threadNum; i++) {
                threads[i].join();
            }

            // 提交事务
            //conn.commit();

        } catch (Exception e) {
            e.printStackTrace();
            try {
                // 出现异常时回滚事务
                conn.rollback();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        } finally {
            try {
                if (ps != null) {
                    ps.close();
                }
                if (conn != null) {
                    conn.close();
                }
                System.out.println("总共耗时：" + (System.currentTimeMillis() - start));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void writeFile() throws IOException {
        long start = System.currentTimeMillis();
        File f2 = new File(filePath);
        FileWriter fw = null;
        BufferedWriter bw = null;
        try {
            fw = new FileWriter(f2);
            bw = new BufferedWriter(fw);
            StringBuffer sb = new StringBuffer();
            for (int i = 0; i < wirteRows; i++) {
                if (i > 0 && i % 10 == 0) {
                    sb.append("\r");
                }
                sb.append(i).append(",").append("tom").append(i).append(",").append(new Random().nextInt(100));
                if (i != wirteRows) {
                    sb.append("|");
                }
                if (i > 0 && i%500000==0) {//每次写入50万
                    bw.write(sb + "\r");
                    sb.setLength(0);//清空
                }
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            bw.close();
            System.out.println("成功写入"+wirteRows+"条,耗时：" + (System.currentTimeMillis() - start));
        }
    }
}
class FileReaderThread implements Runnable {

    private String filePath;
    private int startLine;
    private int endLine;
    private PreparedStatement ps;
    private Connection conn;
    private long miles;
    public FileReaderThread(String filePath, int startLine, int endLine,PreparedStatement ps,Connection conn,long miles) {
        this.filePath = filePath;
        this.startLine = startLine;
        this.endLine = endLine;
        this.ps = ps;
        this.conn = conn;
        this.miles = miles;
    }

    public void run() {
        BufferedReader br = null;
        try {
            br = new BufferedReader(new FileReader(filePath));
            String line = null;
            int lineNumber = 1;
            int insertRows = 0;
            while ((line = br.readLine()) != null) {
                if (lineNumber >= startLine && lineNumber <= endLine) {
                    String[] rowValues = line.split("\\|");
                    for (int i = 0; i < rowValues.length; i++){
                        if(StringUtils.isBlank(rowValues[i])) {
                            continue;
                        }
                        // 处理每一行数据，这里假设每一行都是由逗号分隔的两个值
                        String[] values = rowValues[i].split(",");
                        ps.setString(1, values[0]);
                        ps.setString(2, values[1]);
                        ps.setString(3, values[2]);
                        // 添加到批处理中
                        ps.addBatch();
                        insertRows++;
                    }
                }
                // 5000条提交一次事务
                if(insertRows > 0 && insertRows%100000 == 0){
                    // 执行批处理操作
                    int[] result = ps.executeBatch();
                    System.out.println(System.currentTimeMillis() +",insertRows:" + insertRows + ",result:" + result.length );
                    conn.commit();
                    //3.清空batch
                    ps.clearBatch();
                    insertRows = 0;
                }
                lineNumber++;
            }
            System.out.println(Thread.currentThread().getName() + "正在执行行" + startLine + "," + endLine);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (br != null) {
                    br.close();
                }
                System.out.println(Thread.currentThread().getName() +"耗时：" + (System.currentTimeMillis() - miles));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
